// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tsd.points;

import com.google.common.base.Strings;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.*;
import net.opentsdb.rollup.RollUpDataPoint;

/**
 * 代码优化，提取庞杂的写入代码
 *
 */
public class PutPoint {

    static DataPointType getType(IncomingDataPoint dp){
        if (dp instanceof RollUpDataPoint) {
            return DataPointType.ROLLUP;
        } else if (dp instanceof HistogramPojo) {
            return DataPointType.HISTOGRAM;
        } else {
            return DataPointType.PUT;
        }
    }

    public static Deferred<Object> addPoint(TSDB tsdb, IncomingDataPoint dp){
        AddPointFactory factory = AddPointFactory.getAddPoint(dp);
        AddPoint addPoint = factory.createAddPoint(tsdb ,dp);
        return addPoint.addPoint();
    }

    enum AddPointFactory{
        Histogram{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp) {return new HistogramAddPoint(tsdb, dp);}},
        RollUpInteger{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp) {return new RollUpIntegerAddPoint(tsdb, dp);}},
        RollUpDouble{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp) {return new RollUpDoubleAddPoint(tsdb, dp);}},
        PutInteger{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp) {return new PutIntegerAddPoint(tsdb, dp);}},
        PutDouble{AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp) {return new PutDoubleAddPoint(tsdb, dp);}};

       static AddPointFactory getAddPoint(IncomingDataPoint dp){
           boolean integerFlag = Tags.looksLikeInteger(dp.getValue());

           switch (getType(dp)){
               case HISTOGRAM:
                   return Histogram;
               case PUT:
                   if(integerFlag) return PutInteger;
                   else return PutDouble;
               case ROLLUP:
                   if(integerFlag) return RollUpInteger;
                   else return RollUpDouble;
               default:
                   return null;
           }
       }

       abstract AddPoint createAddPoint(TSDB tsdb, IncomingDataPoint dp);
    }

    interface AddPoint{
        Deferred<Object> addPoint();
    }

    abstract static class AbstractAddPoint implements AddPoint {
        final TSDB tsdb;
        final IncomingDataPoint dp;
        final long timestamp;

        public AbstractAddPoint(TSDB tsdb, IncomingDataPoint dp){
            this.tsdb = tsdb;
            this.dp = dp;
            long time;
            // 时间统一处理为13位
            time = (dp.getTimestamp() & Const.SECOND_MASK)!=0?dp.getTimestamp():dp.getTimestamp()*1000;

            //按照period时间规整
            if(tsdb.getConfig().isEnableRegularTime()){
                long period = tsdb.getConfig().getRegularTimePeriod();
                if(period > 0){
                    time = time - time%(period*1000);
                }
            }
            timestamp = time;
        }
    }

    static class HistogramAddPoint extends AbstractAddPoint{

        public HistogramAddPoint(TSDB tsdb, IncomingDataPoint dp) {
            super(tsdb, dp);
        }

        @Override
        public Deferred<Object> addPoint() {

            final HistogramPojo pojo = (HistogramPojo) dp;

            final Histogram hdp;
            if (Strings.isNullOrEmpty(dp.getValue())) {
                hdp = pojo.toSimpleHistogram(tsdb);
            } else {
                hdp = tsdb.histogramManager().decode(
                        pojo.getId(), pojo.getBytes(), false);
            }
            return tsdb.addHistogramPoint(
                    pojo.getMetric(),
                    timestamp,
                    tsdb.histogramManager().encode(hdp.getId(), hdp, true),
                    pojo.getTags());
        }
    }

    static class RollUpIntegerAddPoint extends AbstractAddPoint {

        public RollUpIntegerAddPoint(TSDB tsdb, IncomingDataPoint dp) {
            super(tsdb, dp);
        }

        @Override
        public Deferred<Object> addPoint() {
            final RollUpDataPoint rdp = (RollUpDataPoint)dp;

            return tsdb.addAggregatePoint(rdp.getMetric(),
                    timestamp,
                    Tags.parseLong(rdp.getValue()),
                    dp.getTags(),
                    rdp.getGroupByAggregator() != null,
                    rdp.getInterval(),
                    rdp.getAggregator(),
                    rdp.getGroupByAggregator());
        }
    }

    static class RollUpDoubleAddPoint extends AbstractAddPoint {

        public RollUpDoubleAddPoint(TSDB tsdb, IncomingDataPoint dp) {
            super(tsdb, dp);
        }

        @Override
        public Deferred<Object> addPoint() {
            final RollUpDataPoint rdp = (RollUpDataPoint)dp;

            return tsdb.addAggregatePoint(rdp.getMetric(),
                    timestamp,
                    (Tags.fitsInFloat(dp.getValue()) ? Float.parseFloat(dp.getValue()) : Double.parseDouble(dp.getValue())),
                    dp.getTags(),
                    rdp.getGroupByAggregator() != null,
                    rdp.getInterval(),
                    rdp.getAggregator(),
                    rdp.getGroupByAggregator());
        }
    }

    static class PutIntegerAddPoint extends AbstractAddPoint {

        public PutIntegerAddPoint(TSDB tsdb, IncomingDataPoint dp) {
            super(tsdb, dp);
        }

        @Override
        public Deferred<Object> addPoint() {
            return tsdb.addPoint(dp.getMetric(),
                    timestamp,
                    Tags.parseLong(dp.getValue()),
                    dp.getTags());
        }
    }

    static class PutDoubleAddPoint extends AbstractAddPoint {

        public PutDoubleAddPoint(TSDB tsdb, IncomingDataPoint dp) {
            super(tsdb, dp);
        }

        @Override
        public Deferred<Object> addPoint() {
            return tsdb.addPoint(dp.getMetric(), timestamp,
                    (Tags.fitsInFloat(dp.getValue()) ?
                            Float.parseFloat(dp.getValue()) :
                            Double.parseDouble(dp.getValue())),
                    dp.getTags());
        }
    }



    public enum DataPointType {
        PUT,
        ROLLUP,
        HISTOGRAM;
    }
}
